package org.niit

import org.apache.spark.{SparkConf, SparkContext}

//统计出每一个省份每个广告被点击数量排行Top3
object RddTest {

  def main(args: Array[String]): Unit = {

    //读取数据
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Test")
    val sc = new SparkContext(sparkConf)
    val dataRDD = sc.textFile("input/agent.log")

    //进行分析
    //1.k - v 分析出来  （ (省份，广告)   ,1） （ (省份，广告)   ,1） （ (省份，广告)   ,1）
    val mappRDD = dataRDD.map(line=>{
      var datas = line.split(" ")
      val pid = datas(1)//省份
      val adid = datas(4)//广告
      ((pid,adid),1 )
    })
    //2.将 （ (省份，广告)   ,1） （ (省份，广告)   ,1） （ (省份，广告)   ,1） 进行聚合 = （ (省份，广告)   ,3）
    val reduceRDD = mappRDD.reduceByKey(_ + _)
    //3. （ (省份，广告)   ,3） ===> (省份，（广告,次数）)
    val newMapRDD = reduceRDD.map{
      case ( (pid,adid),count )=>{
        (pid,(adid,count))
      }
    }
    //4. (省份，（广告,次数）) 根据省份进行分组
    val groupRDD = newMapRDD.groupByKey()
    //5.将分组的数据组内进行降序，取前三
    val resRDD =  groupRDD.mapValues(iter=>{
      iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(3)
    })

    //输出结果
    resRDD.collect().foreach(println)
    sc.stop()
  }

}
